// Package eventbus 全局唯一事件总线
// 职责：
//  1. 异步/优先级出队
//  2. 顺序派发到插件链
//  3. 可选批量持久化
//  4. 优雅关闭、metrics、反压
package event

import (
	"container/heap"
	"context"
	"errors"
	"sync"
	"sync/atomic"
	"time"

	"github.com/sirupsen/logrus"
)

//================== 对外 API ==================

var ErrBusFull = errors.New("eventbus: channel full")

// PriorityEvent 由真实 event.Event 实现
// type PriorityEvent interface {
// 	GetPriority() uint8
// 	SetPriority(uint8)
// }

// Config 启动参数
type Config struct {
	ChanSize          int
	FlushInterval     time.Duration
	MetricsInterval   time.Duration
	EnableMetrics     bool
	EnablePersistence bool
}

// EventBus 核心
type EventBus struct {
	cfg     Config
	mgr     ProcessorManagerInterface
	em      EventManagerInterface
	ch      chan *Event    // 元素是接口
	prioQ   *priorityQueue // nil 表示 FIFO
	stopC   chan struct{}
	wg      sync.WaitGroup
	metrics *metrics

	totalIn  uint64
	totalOut uint64
}

// NewEventBus 创建并启动后台 dispatch
func NewEventBus(cfg Config, mgr ProcessorManagerInterface, em EventManagerInterface) *EventBus {
	if cfg.ChanSize <= 0 {
		cfg.ChanSize = 10240
	}
	if cfg.FlushInterval <= 0 {
		cfg.FlushInterval = 100 * time.Millisecond
	}
	if cfg.MetricsInterval <= 0 {
		cfg.MetricsInterval = 5 * time.Second
	}

	b := &EventBus{
		cfg:     cfg,
		mgr:     mgr,
		em:      em,
		ch:      make(chan *Event, cfg.ChanSize),
		prioQ:   nil,
		stopC:   make(chan struct{}),
		metrics: newMetrics(),
	}
	b.wg.Add(1)
	go b.dispatch()
	return b
}

// Post 异步投递
func (b *EventBus) Post(e *Event) error {
	select {
	case b.ch <- e:
		atomic.AddUint64(&b.totalIn, 1)
		return nil
	default:
		return ErrBusFull
	}
}

// PostSync 同步投递
func (b *EventBus) PostSync(e *Event) error {
	atomic.AddUint64(&b.totalIn, 1)
	b.handleEvent(e)
	return nil
}

// Stop 优雅关闭
func (b *EventBus) Stop(ctx context.Context) error {
	// 1. 通知 dispatch 进入关闭流程
	close(b.stopC)
	// 2. 等待 dispatch 把已入队事件全部处理完
	done := make(chan struct{})
	go func() {
		b.wg.Wait()
		close(done)
	}()
	select {
	case <-done:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}

}

// BusMetrics 快照
type BusMetrics struct {
	TotalIn    uint64
	TotalOut   uint64
	AvgLatency time.Duration
	QueueDepth int
}

// Metrics 返回当前指标
func (b *EventBus) Metrics() BusMetrics {
	return BusMetrics{
		TotalIn:    atomic.LoadUint64(&b.totalIn),
		TotalOut:   atomic.LoadUint64(&b.totalOut),
		AvgLatency: b.metrics.avg(),
		QueueDepth: len(b.ch),
	}
}

// EnablePriority 动态开启优先级堆
func (b *EventBus) EnablePriority() {
	if b.prioQ == nil {
		b.prioQ = &priorityQueue{}
		heap.Init(b.prioQ)
	}
}

//================== 内部实现 ==================

func (b *EventBus) dispatch() {
	defer b.wg.Done()
	tick := time.NewTicker(b.cfg.FlushInterval)
	defer tick.Stop()
	batch := make([]*Event, 0, 200)

	for {
		select {
		case e := <-b.ch:
			// 优先级逻辑
			if b.prioQ != nil {
				heap.Push(b.prioQ, e)
				e = heap.Pop(b.prioQ).(*Event)
			}
			b.handleEvent(e)
			if b.cfg.EnablePersistence {
				batch = append(batch, e)
				if len(batch) >= 200 {
					b.em.BatchAdd(batch)
					batch = batch[:0]
				}
			}

		case <-tick.C:
			if len(batch) > 0 && b.cfg.EnablePersistence {
				b.em.BatchAdd(batch)
				batch = batch[:0]
			}
			if b.cfg.EnableMetrics {
				b.metrics.publish()
			}

		case <-b.stopC:

			// 1. 关闭写入端，让 Post 立即返回错误
			close(b.ch)
			// 2. 把通道里剩余事件全部读完
			for e := range b.ch {
				if b.prioQ != nil {
					heap.Push(b.prioQ, e)
				} else {
					b.handleEvent(e)
				}
			}
			// 3. 优先级堆可能还有
			if b.prioQ != nil {
				for b.prioQ.Len() > 0 {
					e := heap.Pop(b.prioQ).(*Event)
					b.handleEvent(e)
				}
			}
			// 4. 刷持久化缓存
			if len(batch) > 0 && b.cfg.EnablePersistence {
				b.em.BatchAdd(batch)
			}
			return // wg.Done() 执行
		}
	}
}

func (b *EventBus) handleEvent(e *Event) {
	start := time.Now()
	defer func() {
		atomic.AddUint64(&b.totalOut, 1)
		if b.cfg.EnableMetrics {
			b.metrics.record(time.Since(start))
		}
	}()

	// panic 恢复
	defer func() {
		if r := recover(); r != nil {
			logrus.Errorf("eventbus: panic on event: %v", r)
		}
	}()

	b.mgr.Process(e)
}

//================== 优先级堆 ==================

type priorityQueue []*Event

func (pq priorityQueue) Len() int { return len(pq) }
func (pq priorityQueue) Less(i, j int) bool {
	return pq[i].GetPriority() < pq[j].GetPriority() // 0 最高
}
func (pq priorityQueue) Swap(i, j int) { pq[i], pq[j] = pq[j], pq[i] }

func (pq *priorityQueue) Push(x interface{}) {
	*pq = append(*pq, x.(*Event))
}
func (pq *priorityQueue) Pop() interface{} {
	old := *pq
	n := len(old)
	e := old[n-1]
	*pq = old[:n-1]
	return e
}

//================== metrics ==================

type metrics struct {
	sync.Mutex
	window [128]time.Duration
	pos    int
	cnt    int
}

func newMetrics() *metrics { return &metrics{} }

func (m *metrics) record(d time.Duration) {
	m.Lock()
	defer m.Unlock()
	m.window[m.pos] = d
	m.pos = (m.pos + 1) & 127
	if m.cnt < 128 {
		m.cnt++
	}
}

func (m *metrics) avg() time.Duration {
	m.Lock()
	defer m.Unlock()
	if m.cnt == 0 {
		return 0
	}
	var sum time.Duration
	for i := 0; i < m.cnt; i++ {
		sum += m.window[i]
	}
	return sum / time.Duration(m.cnt)
}

func (m *metrics) publish() {
	// logrus.Debugf("eventbus metrics: avgLatency=%v", m.avg())
}

//================== 接口定义 ==================

// EventManagerInterface 由外部 eventManager 实现
type EventManagerInterface interface {
	BatchAdd([]*Event)
}

// ProcessorManagerInterface 由外部 ProcessorManager 实现
type ProcessorManagerInterface interface {
	Process(*Event)
}
